package com.at.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/3/19 9:52 PM
 */
object Spark24_RDD_Req {

  def main(args: Array[String]): Unit = {
    // 环境准备
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("Operator")
    val sc = new SparkContext(sparkConf)

    // TODO agent.log: 时间戳，省份，城市，用户，广告，中间字段使用空格分隔
    //    需求： 统计每一个省份每个广告被点击数量排行的Top2

    // 获取原始数据
    val rdd = sc.textFile("datas/agent.log")
    val dataRDD: RDD[(String, String)] = rdd.map(
      line => {
        val datas = line.split(" ")
        (datas(1), datas(4))
      }
    )

    // 结构转换 ( (省份, 广告), 1)
    val mapRDD = dataRDD.map(t => (t, 1))

    // 将转换结构后的数据进行分组聚合  ( (省份, 广告), sum)
    val reduceRDD = mapRDD.reduceByKey(_ + _)
    //    reduceRDD.collect().foreach(println)

//    val newMapRDD = reduceRDD.map {
//      case ((prv, ad), sum) =>
//        (prv, (ad, sum))
//    }
    val newMapRDD = reduceRDD.map (x => (x._1._1, (x._1._2, x._2)))

    // 将聚合的结果进行结构转换  ( 省份, [(广告A, sumA), (广告B, sumB)])
    //    val groupRDD = reduceRDD.groupBy(t => t._1._1)()
    val groupRDD = newMapRDD.groupByKey()

    groupRDD.collect().foreach(println)
    // 将分组数据降序
    val resultRDD = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(2)
      }
    )

    resultRDD.collect().foreach(println)
    sc.stop()
  }

}
